package com.pisolution.scheduler.runtime.master.scheduler;

import com.pisolution.scheduler.core.graph.Graph;
import com.pisolution.scheduler.core.tuple.Tuple2;
import com.pisolution.scheduler.core.utils.JsonUtils;
import com.pisolution.scheduler.core.utils.OSUtils;
import com.pisolution.scheduler.core.utils.Stopper;
import com.pisolution.scheduler.datasource.instance.ExecutionState;
import com.pisolution.scheduler.datasource.instance.FailureStrategy;
import com.pisolution.scheduler.datasource.instance.Priority;
import com.pisolution.scheduler.datasource.instance.TaskInstance;
import com.pisolution.scheduler.datasource.instance.WorkflowInstance;
import com.pisolution.scheduler.runtime.SchedulerContext;
import com.pisolution.scheduler.runtime.master.MasterConfig;
import com.pisolution.scheduler.runtime.workflow.Flow;
import com.pisolution.scheduler.runtime.workflow.TaskNode;
import com.pisolution.scheduler.runtime.process.ProcessService;

import com.google.common.util.concurrent.ThreadFactoryBuilder;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.annotation.Nullable;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * 负责拆解工作流实例为任务实例并处理执行后的响应信息.
 *
 * @author sora
 * @version 0.1.0
 */
public final class WorkflowScheduler implements Runnable {
    private static final Logger LOGGER = LoggerFactory.getLogger(WorkflowScheduler.class);

    // -------------------------------------------------------------------------------------------------

    /** 用于读取管理节点配置以确认当前是否可以执行工作流实例. */
    private final MasterConfig config;

    /** 待执行的工作流实例. */
    private WorkflowInstance instance;

    /** 用于存储或读取工作流相关数据. */
    private final ProcessService service;

    /** 用于创建任务执行环境. */
    private final ExecutorService executor;

    // -------------------------------------------------------------------------------------------------

    /** 该工作流实例所构成的有向无环图. */
    private Graph<String, TaskNode, Tuple2<String, String>> graph;

    /** 用于发送告警信息. */
    private final AlertManager alertManager;

    /** 确认是否有提交失败的任务. */
    private boolean hasSubmitFailedTask;

    // -------------------------------------------------------------------------------------------------

    /** 正在运行的任务实例. */
    private final Map<TaskScheduler, Future<Boolean>> activeTasks;

    /** 执行失败的任务实例. */
    private final Map<String, TaskInstance> errorTasks;

    /** 执行完成的任务实例. */
    private final Map<String, TaskInstance> completeTasks;

    /** 前序节点失败的任务实例. */
    private final Map<String, TaskInstance> dependFailedTasks;

    /** 需要容错的任务实例. */
    private final List<TaskInstance> tolerantTasks;

    /** 准备就绪的任务实例 (随时可供执行). */
    private final Queue<TaskInstance> readyTasks;

    // -------------------------------------------------------------------------------------------------
    //  初始化
    // -------------------------------------------------------------------------------------------------

    /**
     * 初始化 {@link WorkflowScheduler}.
     *
     * @param instance 待执行工作流实例.
     * @param service 用于存储或读取工作流相关数据.
     */
    public WorkflowScheduler(final WorkflowInstance instance, final ProcessService service) {
        this.instance = instance;
        this.service = service;
        this.config = SchedulerContext.get(MasterConfig.class);
        this.executor = Executors.newFixedThreadPool(
            this.config.getMasterSchedulerThreads(),
            new ThreadFactoryBuilder().setDaemon(true).setNameFormat("master-workflow-execution-thread").build()
        );

        this.alertManager = new AlertManager();
        this.hasSubmitFailedTask = false;

        this.activeTasks = new ConcurrentHashMap<>();
        this.errorTasks = new ConcurrentHashMap<>();
        this.completeTasks = new ConcurrentHashMap<>();
        this.dependFailedTasks = new ConcurrentHashMap<>();
        this.tolerantTasks = new ArrayList<>();
        this.readyTasks = new LinkedBlockingQueue<>();
    }

    // -------------------------------------------------------------------------------------------------
    //  处理工作流实例
    // -------------------------------------------------------------------------------------------------

    @Override
    public void run() {
        try {
            // 1. 确认待执行工作流实例是否有效.
            if (this.instance == null) {
                LOGGER.info("The workflow instance is not exist.");

                return;
            }

            // 2. 确认待执行工作流实例是否已经执行完成.
            if (this.instance.getState().isFinished()) {
                LOGGER.info("The workflow instance {} is done.", this.instance.getName());

                return;
            }

            // 3. 准备工作流实例执行环境.
            this.prepare();

            // 4. 执行工作流实例.
            this.execute();

            // 5. 更新工作流实例的运行结束时间.
            this.instance.setEndTime(new Date());
            this.service.saveWorkflowInstance(this.instance);

            // 6. 如果当前工作流实例处于等待线程的状态 则创建恢复等待的指令.
            if (this.instance.getState() == ExecutionState.WAITING_THREAD) {
                this.service.createRecoveryWaitingThreadCommand(null, this.instance);
            }

            // 7. 依据工作流实例的执行状态 发送相应的告警信息.
            this.alertManager.sendWorkflowInstanceAlert(
                this.instance,
                this.service.findTaskInstancesByWorkflowInstanceId(this.instance.getId())
            );
        } catch (Exception e) {
            LOGGER.error("The execution of the workflow instance {} failed.", this.instance.getName(), e);

            this.instance.setState(ExecutionState.FAILURE);
            this.instance.setEndTime(new Date());
            this.service.saveWorkflowInstance(this.instance);
        } finally {
            this.executor.shutdown();
        }
    }

    // -------------------------------------------------------------------------------------------------
    //  准备阶段
    // -------------------------------------------------------------------------------------------------

    /**
     * 准备工作流实例执行环境 包括以下步骤.
     *
     * <ul>
     *     <li>将工作流实例对象转换为可执行有向无环图</li>
     *     <li>初始化任务存储队列</li>
     * </ul>
     */
    private void prepare() {
        // 1. 从工作流实例中读取其原始数据结构并生成图.
        this.graph = this.generateGraph(this.instance.getContent());

        // 2. 确认构建出来的图是否有效.
        if (this.graph == null) {
            LOGGER.error("The graph generated from the workflow instance {} is invalid.", this.instance.getName());

            return;
        }

        // 3. 初始化任务队列.
        this.errorTasks.clear();
        this.completeTasks.clear();
        this.dependFailedTasks.clear();

        // 4. 查询当前工作流实例下所有的任务实例.
        final List<TaskInstance> tasks = this.service.findTaskInstancesByWorkflowInstanceId(this.instance.getId());

        // 5. 根据任务实例的当前状态 填充相应任务队列.
        for (final TaskInstance task: tasks) {
            if (task.isTaskComplete()) {
                this.completeTasks.put(task.getName(), task);

                continue;
            }

            if (task.isTaskError()) {
                this.errorTasks.put(task.getName(), task);
            }
        }

        LOGGER.info("The preparation of the execution of the workflow instance {} is completed.", this.instance.getName());
    }

    /**
     * 解析工作流实例中的任务节点定义内容并以此构建出可供执行的有向无环图.
     *
     * @param content 包含工作流实例中所有任务节点的定义内容.
     *
     * @return 从原始定义内容中生成的可供执行的有向无环图.
     */
    @Nullable
    private Graph<String, TaskNode, Tuple2<String, String>> generateGraph(final String content) {
        try {
            // 1. 将原始工作流定义内容转换为实体对象并获取其中的任务节点.
            final List<TaskNode> tasks = ((Flow) JsonUtils.deserialize(content, Flow.class)).getTasks();
            final Graph<String, TaskNode, Tuple2<String, String>> graph = new Graph<>();

            // 2. 添加图中的节点.
            for (final TaskNode node: tasks) {
                graph.addNode(node.getName(), node);
            }

            // 3. 添加图中的边.
            for (final TaskNode node: tasks) {
                node.getConnections().values().forEach(source -> graph.addEdge(source.getName(), node.getName()));
            }

            return graph;
        } catch (Exception e) {
            LOGGER.error("Could not generate the graph from the given workflow instance {}.", this.instance.getName());

            return null;
        }
    }

    // -------------------------------------------------------------------------------------------------
    //  运行阶段
    // -------------------------------------------------------------------------------------------------

    /**
     * 执行工作流实例.
     */
    @SuppressWarnings("BusyWait")
    private void execute() {
        // 1. 寻找图中下一个可供执行的任务实例并追加到准备就绪的任务实例队列中.
        this.resolveNextTask(null);

        // 2. 如果工作流实例尚未执行完成 则一直等待其执行完成.
        boolean canSendWarningAlert = true;

        while (!this.instance.getState().isFinished() && Stopper.isRunning()) {
            // 3. 校验当前执行的工作流实例是否已经超时 (只发送一次告警信息).
            if (canSendWarningAlert && this.checkInstanceTimeout()) {
                this.alertManager.sendWorkflowInstanceTimeoutAlert(this.instance);

                canSendWarningAlert = false;
            }

            // 4. 处理当前正在运行的任务实例.
            for (final Map.Entry<TaskScheduler, Future<Boolean>> entry: this.activeTasks.entrySet()) {
                // 5. 如果当前任务节点尚未运行完成 则直接跳过.
                if (!entry.getValue().isDone()) {
                    continue;
                }

                // 6. 从数据库中寻找当前任务实例.
                final TaskInstance task = this.service.findTaskInstanceById(entry.getKey().getTask().getId());

                // 7. 如果当前任务实例为空 则判定为提交任务实例至工作节点失败.
                if (task == null) {
                    this.hasSubmitFailedTask = true;
                    this.activeTasks.remove(entry.getKey());

                    continue;
                }

                // 8. 如果任务实例已经执行完成 则从当前执行的任务实例中移除该任务实例.
                if (task.getState().isFinished()) {
                    this.activeTasks.remove(entry.getKey());
                }

                LOGGER.info("The execution of the task {} is completed with state {}.", task.getName(), task.getState());

                // 9. 如果执行成功了 则追加到已经成功的任务实例集合中并寻找下一个可执行任务实例.
                if (task.getState() == ExecutionState.SUCCESS) {
                    this.completeTasks.put(task.getName(), task);
                    this.resolveNextTask(task.getName());

                    continue;
                }

                // 10. 如果因为节点容错而导致执行失败 则追加到容错任务列表中供后续告警使用.
                if (task.getState() == ExecutionState.NEED_FAULT_TOLERANCE) {
                    this.tolerantTasks.add(task);
                }

                // 11. 如果执行失败了 则进一步判定是否可以进行重试.
                if (task.getState() == ExecutionState.FAILURE) {
                    if (task.canRetry()) {
                        this.readyTasks.add(task);

                        continue;
                    }

                    this.completeTasks.put(task.getName(), task);
                    this.errorTasks.put(task.getName(), task);

                    // 12. 如果工作流实例中失败策略为节点失败即停止运行 则一并取消其他正在运行的任务.
                    if (this.instance.getStrategy() == FailureStrategy.END) {
                        this.killRunningTasks();
                    }

                    continue;
                }

                this.completeTasks.put(task.getName(), task);
            }

            // 13. 如果有需要容错的任务 则发送相关告警信息并清空该列表.
            if (!this.tolerantTasks.isEmpty()) {
                this.alertManager.sendWorkerServerFaultTolerantAlert(this.instance, this.tolerantTasks);
                this.tolerantTasks.clear();
            }

            // 14. 如果有任务失败 其他暂停运行的任务需要被标记为死亡状态.
            if (!this.errorTasks.isEmpty()) {
                for (Map.Entry<String, TaskInstance> entry: this.completeTasks.entrySet()) {
                    final TaskInstance task = entry.getValue();

                    if (task.getState() == ExecutionState.PAUSE) {
                        task.setState(ExecutionState.KILL);

                        this.completeTasks.put(entry.getKey(), task);
                        this.service.saveTaskInstance(task);
                    }
                }
            }

            // 15. 如果当前可用资源充足 则尝试处理待执行任务.
            if (OSUtils.availablePhysicalMemory() >= this.config.getReservedMemory() && OSUtils.loadAverage() <= this.config.getMaxCPULoadAvg()) {
                this.resolveStandByTask();
            }

            // 16. 以固定间隔执行此循环并更新工作流实例状态.
            try {
                Thread.sleep(1000L);
            } catch (InterruptedException e) {
                LOGGER.error(e.getMessage(), e);
            }

            this.updateExecutionState();
        }

        LOGGER.info("The workflow instance {} is completed with state {}.", this.instance.getName(), this.instance.getState());
    }

    /**
     * 预处理下一个可供执行的任务节点.
     *
     * @param taskName 当前已经执行完成的任务节点名称.
     */
    private void resolveNextTask(@Nullable final String taskName) {
        // 1. 获取有向无环图中下一个可供执行的任务节点.
        final Set<String> nextTasks = this.findNextTask(taskName);
        final List<TaskInstance> tasks = new ArrayList<>();

        // 2. 将可供执行的任务节点转换为任务实例.
        for (final String task: nextTasks) {
            tasks.add(this.createTask(task));
        }

        // 3. 确认每一个任务是否准备就绪.
        for (final TaskInstance task: tasks) {
            // 4. 如果已经被其他节点添加到待执行任务节点队列中 则不予处理该任务节点.
            if (this.readyTasks.contains(task)) {
                continue;
            }

            // 5. 如果已经处理完成 则不予处理该任务节点.
            if (this.completeTasks.containsKey(task.getName())) {
                continue;
            }

            // 6. 如果尚未收到该任务节点的暂停或取消指令 则追加到待处理任务节点队列中.
            if (task.getState() == ExecutionState.PAUSE || task.getState() == ExecutionState.KILL || task.getState() == ExecutionState.STOP) {
                continue;
            }

            this.readyTasks.add(task);
        }
    }

    /**
     * 从计算图中寻找下一个可以执行的节点.
     *
     * @param name 当前已经执行完成的节点名称.
     *
     * @return 可供执行的节点集合.
     */
    private Set<String> findNextTask(@Nullable final String name) {
        // 1. 如果未指定节点名称 则判定下一个可供执行的节点为根节点.
        final Collection<String> tasks = name == null ? this.graph.getStartNode() : this.graph.getSubsequentNodes(name);
        final Set<String> availableTasks = new HashSet<>();

        for (final String task: tasks) {
            // 2. 获取节点中存储的数据.
            final TaskNode node = this.graph.getNode(task);

            // 3. 如果该节点的所有前序依赖节点尚未执行完成 则该节点尚不可执行.
            if (!CollectionUtils.isSubCollection(node.getDependentNodes(), this.completeTasks.keySet())) {
                continue;
            }

            // 4. 如果该节点已经执行完成 则寻找该节点下的可供执行子节点.
            if (this.completeTasks.containsKey(task)) {
                availableTasks.addAll(this.findNextTask(task));

                continue;
            }

            // 5. 追加该节点到可执行节点集合中.
            availableTasks.add(task);
        }

        return availableTasks;
    }

    /**
     * 将给定的任务节点名称转换为任务实例对象.
     *
     * @param taskName 待执行任务节点名称.
     *
     * @return 可执行任务实例对象.
     */
    private TaskInstance createTask(final String taskName) {
        // 1. 确认待转换任务实例是否已经存在.
        final Optional<TaskInstance> task = this.service
            .findTaskInstancesByWorkflowInstanceId(this.instance.getId())
            .stream()
            .filter(item -> item.getName().equals(taskName))
            .findFirst();

        // 2. 如果任务实例已经存在 则无需转换.
        if (task.isPresent()) {
            return task.get();
        }

        // 3. 创建新的任务实例.
        final TaskInstance instance = new TaskInstance();
        final TaskNode node = this.graph.getNode(taskName);

        instance.setName(taskName);
        instance.setType(node.getType());
        instance.setPriority(node.getPriority() == null ? Priority.MEDIUM : node.getPriority());
        instance.setContent(JsonUtils.serializeToString(node));
        instance.setWorkflowId(this.instance.getWorkflowId());
        instance.setWorkflowInstanceId(this.instance.getId());
        instance.setUserId(this.instance.getUserId());
        instance.setWorkerGroup(this.instance.getWorkerGroup());
        instance.setMaxRetryCount(node.getMaxRetryCount());
        instance.setRetryInterval(node.getInterval());
        instance.setState(ExecutionState.SUBMITTED_SUCCESS);
        instance.setStartTime(null);
        instance.setRetryCount(0);

        return instance;
    }

    /**
     * 确认当前工作流实例是否已经执行超时.
     *
     * @return 如果超时了则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean checkInstanceTimeout() {
        // 1. 如果未设置超时属性 则判定为未超时.
        if (this.instance.getTimeout() == 0) {
            return false;
        }

        // 2. 确认启动时间与当前时间之差是否已经超过超时时间.
        return DateUtils.addMinutes(this.instance.getStartTime(), this.instance.getTimeout()).before(new Date());
    }

    /**
     * 强制相关运行中的任务全部退出.
     */
    private void killRunningTasks() {
        for (final Map.Entry<TaskScheduler, Future<Boolean>> entry: this.activeTasks.entrySet()) {
            // 1. 如果待退出的任务实例已经完成 则无需处理.
            final TaskInstance task = this.service.findTaskInstanceById(entry.getKey().getTask().getId());

            if (task != null && task.getState().isFinished()) {
                continue;
            }

            // 2. 确认待退出的节点是否还未完成逻辑执行.
            if (!entry.getValue().isDone()) {
                LOGGER.info(
                    "Force killing the task instance {} of the workflow instance {}.",
                    entry.getKey().getTask().getName(),
                    this.instance.getName()
                );

                entry.getKey().kill();
            }
        }
    }

    /**
     * 尝试将准备就绪的任务节点追加到待运行任务列表中 如果该节点上游的节点并未准备就绪 则无需处理.
     */
    private void resolveStandByTask() {
        try {
            while (!this.readyTasks.isEmpty()) {
                // 1. 获取已经准备就绪的任务节点.
                final TaskInstance task = this.readyTasks.peek();

                // 2. 确认该任务节点的前序依赖节点是否已经执行成功.
                final DependentState state = this.getDependentTaskState(task.getName());

                // 3. 由于给定任务节点有重试的可能性 需判定该节点是否已经超过了重试时间.
                if (state == DependentState.SUCCESS) {
                    if (this.canTaskContinueRetrying(task)) {
                        // 4. 创建任务节点调度线程以调度此任务.
                        final TaskScheduler scheduler = new TaskScheduler(task);

                        this.activeTasks.putIfAbsent(scheduler, this.executor.submit(scheduler));
                        this.readyTasks.remove(task);

                        LOGGER.info("Submit the task instance {} to the task scheduler.", task.getName());
                    }
                }

                // 5. 前序节点执行失败 故不予提交执行此任务节点并从待执行任务节点队列中予以移除.
                if (state == DependentState.FAILURE) {
                    this.dependFailedTasks.put(task.getName(), task);
                    this.readyTasks.remove(task);
                }
            }
        } catch (Exception e) {
            LOGGER.error("Could not resolve the stand by task.", e);
        }
    }

    /**
     * 通过给定节点名称查询相应节点信息并获取其前序依赖节点的执行状态.
     *
     * @param taskName 待查询节点名称.
     *
     * @return 待查询节点的前序依赖节点的执行状态.
     */
    private DependentState getDependentTaskState(final String taskName) {
        // 1. 如果当前任务是图中的顶点 则无前序依赖节点 判定前序节点执行成功.
        if (this.graph.getStartNode().contains(taskName)) {
            return DependentState.SUCCESS;
        }

        // 2. 获取当前节点的所有前序依赖的节点.
        final List<String> nodes = this.graph.getNode(taskName).getDependentNodes();

        for (final String node: nodes) {
            // 3. 如果前序节点尚未完成 则判定前序节点尚在执行过程中.
            if (!this.completeTasks.containsKey(node)) {
                return DependentState.WAITING;
            }

            // 4. 如果前序节点处于暂停运行或取消运行状态 则判定为尚在执行过程中.
            final ExecutionState state = this.completeTasks.get(node).getState();

            if (state == ExecutionState.PAUSE || state == ExecutionState.KILL || state == ExecutionState.STOP) {
                return DependentState.WAITING;
            }

            // 5. 如果前序节点处于运行失败的状态 则判定前序节点为执行失败.
            if (state == ExecutionState.FAILURE || state == ExecutionState.NEED_FAULT_TOLERANCE) {
                return DependentState.FAILURE;
            }
        }

        return DependentState.SUCCESS;
    }

    /**
     * 确认给定任务实例是否可以继续重试运行.
     *
     * @param task 待确认任务实例.
     *
     * @return 如果可以继续重试运行 则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean canTaskContinueRetrying(final TaskInstance task) {
        // 1. 如果当前节点并不处于失败状态 则允许重试运行 (针对尚未运行过的任务实例).
        if (task.getState() != ExecutionState.FAILURE) {
            return true;
        }

        // 2. 如果是新创建的任务实例或者任务尚未指定重试时间与此处 则允许重试运行.
        if (task.getId() == 0 || task.getMaxRetryCount() == 0 || task.getRetryInterval() == 0) {
            return true;
        }

        // 3. 如果尚未处于可重试时间间隔之内 则允许重试运行 (任务节点已经运行过至少一次).
        return DateUtils.addSeconds(task.getEndTime(), task.getRetryInterval() * 60).before(new Date());
    }

    /**
     * 更新工作流实例的执行状态.
     */
    private void updateExecutionState() {
        // 1. 获取最新的运行状态.
        final ExecutionState state = this.getLatestExecutionState();

        if (this.instance.getState() != state) {
            LOGGER.info(
                "The execution state of the workflow instance {} changed from {} to {}.",
                this.instance.getName(),
                this.instance.getState(),
                state
            );

            // 2. 获取最新的工作流实例对象.
            final WorkflowInstance instance = this.service.findWorkflowInstanceById(this.instance.getId());

            // 3. 更改工作流实例运行状态并覆盖原有工作流实例对象.
            instance.setState(state);
            instance.setWorkflow(this.instance.getWorkflow());

            this.service.saveWorkflowInstance(instance);
            this.instance = instance;
        }
    }

    /**
     * 获取调度中的工作流实例的最新运行状态.
     *
     * @return 最新运行状态.
     */
    private ExecutionState getLatestExecutionState() {
        // 1. 查询最新的工作流实例运行状态.
        final WorkflowInstance instance = this.service.findWorkflowInstanceById(this.instance.getId());
        final ExecutionState state = instance.getState();

        // 2. 如果有正在运行的任务节点或尚未提交运行的重试节点 则进一步判定是否正在运行.
        if (!this.activeTasks.isEmpty() || this.hasRetryTask()) {
            if (state == ExecutionState.READY_STOP || state == ExecutionState.READY_PAUSE || state == ExecutionState.WAITING_THREAD) {
                return state;
            }

            return ExecutionState.RUNNING;
        }

        // 3. 如果工作流实例执行失败 则判定为运行失败.
        if (this.isExecuteFailed()) {
            return ExecutionState.FAILURE;
        }

        // 4. 如果有处于等待线程状态的任务节点 则判定为等待线程状态.
        if (this.completeTasks.values().stream().anyMatch(task -> task.getState() == ExecutionState.WAITING_THREAD)) {
            return ExecutionState.WAITING_THREAD;
        }

        // 5. 如果处于等待暂停状态 则进一步细化判定条件.
        if (state == ExecutionState.READY_PAUSE) {
            if (this.hasRetryTask()) {
                return ExecutionState.FAILURE;
            }

            if (this.completeTasks.values().stream().anyMatch(task -> task.getState() == ExecutionState.PAUSE) || !this.readyTasks.isEmpty()) {
                return ExecutionState.PAUSE;
            }

            return ExecutionState.SUCCESS;
        }

        // 6. 如果处于等待停止状态 则进一步细化判定条件.
        if (state == ExecutionState.READY_STOP) {
            if (this.completeTasks.values().stream().anyMatch(task -> task.getState() == ExecutionState.STOP || task.getState() == ExecutionState.KILL)) {
                return ExecutionState.STOP;
            }

            return ExecutionState.SUCCESS;
        }

        // 7. 如果处于运行中的状态 则需要判定强制退出的任务数量等条件.
        if (state == ExecutionState.RUNNING) {
            if (!this.readyTasks.isEmpty()) {
                return ExecutionState.RUNNING;
            }

            if (this.completeTasks.values().stream().anyMatch(task -> task.getState() == ExecutionState.KILL)) {
                return ExecutionState.FAILURE;
            }

            return ExecutionState.SUCCESS;
        }

        return state;
    }

    /**
     * 确认准备提交执行的任务节点中是否有节点非第一次运行.
     *
     * @return 如果有非第一次运行的节点 则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean hasRetryTask() {
        for (final TaskInstance task : this.readyTasks) {
            final ExecutionState state = task.getState();

            if (state == ExecutionState.FAILURE || state == ExecutionState.NEED_FAULT_TOLERANCE) {
                return true;
            }
        }

        return false;
    }

    /**
     * 确认当前工作流实例是否执行失败.
     *
     * @return 如果执行失败则返回 {@code true} 否则返回 {@code false}.
     */
    private boolean isExecuteFailed() {
        if (this.hasSubmitFailedTask || !this.errorTasks.isEmpty() || !this.dependFailedTasks.isEmpty()) {
            if (this.instance.getStrategy() == FailureStrategy.END) {
                return true;
            }

            return this.readyTasks.isEmpty() || this.activeTasks.isEmpty();
        }

        return false;
    }
}
